package com.imooc.socialecom.runner;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.google.protobuf.InvalidProtocolBufferException;
import com.imooc.socialecom.pojo.Item;
import com.imooc.socialecom.service.ItemService;
import org.apache.http.HttpHost;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestHighLevelClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Component;

import java.net.InetSocketAddress;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

@Component
public class CanalApplicationRunner implements ApplicationRunner {
    private static final String SERVER_ADDRESS = "127.0.0.1";
    private static final Integer PORT = 1111;
    private static final String DESTINATION = "example";
    private static final String USERNAME = "";
    private static final String PASSWORD = "";

    @Autowired
    private ItemService itemService;

    private RestHighLevelClient client = null;

    @Override
    public void run(ApplicationArguments args) throws Exception {
        client = new RestHighLevelClient(RestClient.builder(new HttpHost("localhost", 9200, "http")));

        CanalConnector canalConnector = CanalConnectors.newSingleConnector(new InetSocketAddress(
                SERVER_ADDRESS,
                PORT
        ), DESTINATION, USERNAME, PASSWORD);
        canalConnector.connect();

        canalConnector.subscribe(".*\\..*");

        canalConnector.rollback();

        while (true) {
            Message message = canalConnector.getWithoutAck(100);
            long batchId = message.getId();
            if (batchId != -1) {
                List<CanalEntry.Entry> entries = message.getEntries();
                for (CanalEntry.Entry entry : entries) {
                    if (entry.getEntryType() != CanalEntry.EntryType.ROWDATA) {
                        continue;
                    }
                    try {
                        String tableName = entry.getHeader().getTableName();
                        if (!tableName.equals("item")) {
                            continue;
                        }
                        CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
                        for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
                            List<CanalEntry.Column> afterColumnsList = null;
                            switch (rowChange.getEventType()) {
                                case INSERT:
                                    afterColumnsList = rowData.getAfterColumnsList();
                                    break;
                                case UPDATE:
                                    afterColumnsList = rowData.getAfterColumnsList();
                                    break;

                                default:
                                    break;
                            }
                            for (CanalEntry.Column c : afterColumnsList) {
                                if (c.getName().equals("id")) {
                                    Item item = itemService.getById(c.getValue());
                                    if (item == null) {
                                        continue;
                                    }
                                    // 更新ElasticSearch中的数据
                                    Map<String, Object> map = new HashMap<>();
                                    map.put("name", item.getName());
                                    client.index(new IndexRequest("item").id(item.getId().toString()).source(map), RequestOptions.DEFAULT);
                                }
                            }
                        }


                    } catch (InvalidProtocolBufferException ex) {
                        ex.printStackTrace();
                    }
                }

                canalConnector.ack(batchId);
            }
        }
    }
}
